package com.xjrh.traintravel.collector.mq;

import com.xjrh.traintravel.collector.mq.msg.*;
import com.xjrh.traintravel.collector.netty.command.front.VersionData;
import com.xjrh.traintravel.collector.netty.command.mvb.MvbData;
import com.xjrh.traintravel.collector.netty.command.mvb.MvbStation;
import com.xjrh.traintravel.collector.netty.command.station.TravelStationRegionData;
import lombok.extern.slf4j.Slf4j;
import com.alibaba.fastjson.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * Kafka消息生产者
 *
 * @author snps
 * @create 2022-03-30 14:41
 * @describe
 **/
@Component
@Slf4j
public class KafkaProducer {

    // 采集数据 Topic
    static final String TOPIC_COLLECTION_DATA = "topic.collection.data";

    // 波磨数据 Topic
    static final String TOPIC_WAVE_DATA = "topic.wave.data";

    // 故障数据 Topic
    static final String TOPIC_FAULT = "topic.fault.data";

    // MVB数据 Topic
    static final String TOPIC_MVB_DATA = "topic.mvb.data";

    // MVB站点 Topic
    static final String TOPIC_MVB_STATION = "topic.mvb.station";

    // 获取主机（软件、硬件）版本
    static final String TOPIC_EQUIPMENT_VERSION = "topic.equipment.version";

    // 设备异常信息 Topic
    static final String TOPIC_EQUIPMENT_STATUS = "topic.equipment.status";


    // MVB文件数据 Topic
    static final String TOPIC_MVB_FILE_DATA = "topic.mvb.file-data";

    // 算法检测的部件故障 Topic
    static final String TOPIC_PART_FAULT = "topic.part.fault";

    // 列车运行区间信息 Topic
    static final String TOPIC_TRAVEL_STATION_REGION = "topic.travel.station.region";


    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;


    /**
     * 发送MVB数据
     *
     * @param mvbData
     */
    public void sendMvbData(MvbData mvbData) {
        String obj2String = JSONObject.toJSONString(mvbData);
        // log.warn("MVB-要发送的消息：{}", obj2String);

        // 发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_MVB_DATA, obj2String);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                //发送失败的处理
                log.error(TOPIC_MVB_DATA + " - 发送消息失败：" + throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                //成功的处理
                // log.info(TOPIC_MVB_DATA + " - 发送消息成功：" + stringObjectSendResult.toString());
            }
        });
    }

    /**
     * 发送采集数据
     *
     * @param collectionDataMsg
     */
    public void sendCollectionData(CollectionDataMsg collectionDataMsg) {
        String obj2String = JSONObject.toJSONString(collectionDataMsg);
        // log.warn("采集数据-要发送的消息：{}", obj2String);

        // 发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_COLLECTION_DATA, obj2String);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                // 发送失败的处理
                log.error(TOPIC_COLLECTION_DATA + " - 发送消息失败：" + throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                // 成功的处理
                // log.info(TOPIC_COLLECTION_DATA + " - 发送消息成功：" + stringObjectSendResult.toString());
            }
        });
    }

    /**
     * 发送波磨数据
     *
     * @param waveDataMsg
     */
    public void sendWaveData(WaveDataMsg waveDataMsg) {
        String obj2String = JSONObject.toJSONString(waveDataMsg);
        // log.warn("波磨数据-要发送的消息：{}", obj2String);

        // 发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_WAVE_DATA, obj2String);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                // 发送失败的处理
                log.error(TOPIC_WAVE_DATA + " - 发送消息失败：" + throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                // 成功的处理
                // log.info(TOPIC_WAVE_DATA + " - 发送消息成功：" + stringObjectSendResult.toString());
            }
        });
    }

    /**
     * 发送MVB数据
     *
     * @param mvbDataMsg
     */
    public void sendMvbFileData(MvbDataMsg mvbDataMsg) {
        String obj2String = JSONObject.toJSONString(mvbDataMsg);
        // log.warn("故障数据-要发送的消息：{}", obj2String);

        // 发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_MVB_FILE_DATA, obj2String);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                // 发送失败的处理
                log.error(TOPIC_MVB_FILE_DATA + " - 发送消息失败：" + throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                // 成功的处理
                // log.info(TOPIC_MVB_FILE_DATA + " - 发送消息成功：" + stringObjectSendResult.toString());
            }
        });
    }

    /**
     * 发送：主机（软件、硬件）版本信息
     *
     * @param versionData
     */
    public void sendEquipmentVersion(VersionData versionData) {
        String obj2String = JSONObject.toJSONString(versionData);

        // 发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_EQUIPMENT_VERSION, obj2String);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                // 发送失败的处理
                log.error(TOPIC_EQUIPMENT_VERSION + " - 发送消息失败：" + throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                // 成功的处理
                // log.info(TOPIC_EQUIPMENT_VERSION + " - 发送消息成功：" + stringObjectSendResult.toString());
            }
        });
    }

    /**
     * 发送：设备异常信息
     *
     * @param equipmentExceptionMsg
     */
    public void sendEquipmentFault(EquipmentExceptionMsg equipmentExceptionMsg) {
        String obj2String = JSONObject.toJSONString(equipmentExceptionMsg);
        // log.warn("故障数据-要发送的消息：{}", obj2String);

        // 发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_EQUIPMENT_STATUS, obj2String);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                // 发送失败的处理
                log.error(TOPIC_EQUIPMENT_STATUS + " - 发送消息失败：" + throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                // 成功的处理
                // log.info(TOPIC_EQUIPMENT_STATUS + " - 发送消息成功：" + stringObjectSendResult.toString());
            }
        });
    }

    /**
     * 发送：Mvb站点信息
     *
     * @param mvbStation
     */
    public void sendMvbStation(MvbStation mvbStation) {
        String obj2String = JSONObject.toJSONString(mvbStation);
        // log.warn("MVB-要发送的消息：{}", obj2String);

        // 发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_MVB_STATION, obj2String);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                //发送失败的处理
                log.error(TOPIC_MVB_STATION + " - 发送消息失败：" + throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                //成功的处理
                // log.info(TOPIC_MVB_STATION + " - 发送消息成功：" + stringObjectSendResult.toString());
            }
        });
    }


    /**
     * 发送：算法检测到的部件故障
     *
     * @param partFaultMsg
     */
    public void sendPartFault(PartFaultMsg partFaultMsg) {
        String obj2String = JSONObject.toJSONString(partFaultMsg);
        // log.warn("故障数据-要发送的消息：{}", obj2String);

        // 发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_PART_FAULT, obj2String);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                // 发送失败的处理
                log.error(TOPIC_PART_FAULT + " - 发送消息失败：" + throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                // 成功的处理
                // log.info(TOPIC_FAULT + " - 发送消息成功：" + stringObjectSendResult.toString());
            }
        });
    }

    /**
     * 发送：列车运行区间信息
     *
     * @param travelStationRegionData
     */
    public void sendTravelStationRegion(TravelStationRegionData travelStationRegionData) {
        String obj2String = JSONObject.toJSONString(travelStationRegionData);
        // log.warn("列车运行区间信息-要发送的消息：{}", obj2String);

        // 发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_TRAVEL_STATION_REGION, obj2String);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                // 发送失败的处理
                log.error(TOPIC_TRAVEL_STATION_REGION + " - 发送消息失败：" + throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                // 成功的处理
                // log.info(TOPIC_TRAVEL_STATION_REGION + " - 发送消息成功：" + stringObjectSendResult.toString());
            }
        });
    }


}